package com.heima.behavior.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.behavior.entity.ApReadBehavior;
import com.heima.behavior.mapper.ApReadBehaviorMapper;
import com.heima.behavior.properties.KafkaProperties;
import com.heima.behavior.service.ApReadBehaviorService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.threadlocal.UserThreadLocalUtils;
import com.heima.common.util.JsonUtils;
import com.heima.feign.ApArticleFeignClient;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.behavior.dtos.ReadBehaviorDto;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Properties;

/**
 * <p>
 * APP阅读行为表 服务实现类
 * </p>
 *
 * @author HM
 * @since 2023-04-09
 */
@Service
public class ApReadBehaviorServiceImpl extends ServiceImpl<ApReadBehaviorMapper, ApReadBehavior> implements ApReadBehaviorService {
    @Autowired
    private KafkaProducer kafkaProducer;

    /**
     * 阅读行为
     *
     * @param dto
     */
    @Override
    public void readBehavior(ReadBehaviorDto dto) {
        Long articleId = dto.getArticleId();
        LambdaQueryWrapper<ApReadBehavior> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ApReadBehavior::getArticleId, articleId);
        queryWrapper.eq(ApReadBehavior::getApUserId,UserThreadLocalUtils.getUserId());
        ApReadBehavior readBehavior = this.getOne(queryWrapper);
        if (readBehavior == null) {
            ApReadBehavior apReadBehavior = new ApReadBehavior();
            apReadBehavior.setArticleId(dto.getArticleId());
            apReadBehavior.setApUserId(UserThreadLocalUtils.getUserId());
            apReadBehavior.setCount(dto.getCount());
            apReadBehavior.setCreatedTime(new Date());
            apReadBehavior.setUpdatedTime(new Date());
            this.save(apReadBehavior);
        } else {
            readBehavior.setCount(readBehavior.getCount() + dto.getCount());
            readBehavior.setUpdatedTime(new Date());
            readBehavior.setApUserId(UserThreadLocalUtils.getUserId());
            this.updateById(readBehavior);
        }
        sendMsgToKafka(dto);
    }

    /**
     * 往kafka发送消息
     * @param dto
     */
    private void sendMsgToKafka(ReadBehaviorDto dto) {
        ArticleBehaviorMsg msg = new ArticleBehaviorMsg();
        msg.setArticleId(dto.getArticleId());
        msg.setNum(dto.getCount());
        msg.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JsonUtils.toString(msg));
        kafkaProducer.send(record);
        System.out.println("发送数据" + JsonUtils.toString(msg));
    }


}
